package com.waves.task.strategy.receive;

import com.waves.core.constant.TimeConstant;
import com.waves.core.constant.redis.TaskRedisKey;
import com.waves.core.service.LockService;
import com.waves.core.utils.AssertUtil;
import com.waves.core.utils.CompletableFutureUtils;
import com.waves.core.utils.LoginUtils;
import com.waves.core.utils.RedisUtils;
import com.waves.core.web.domain.R;
import com.waves.core.web.domain.entities.User;
import com.waves.task.dao.TaskDao;
import com.waves.task.dao.TaskReceiveDao;
import com.waves.task.domain.entity.Task;
import com.waves.task.domain.entity.TaskReceive;
import com.waves.task.domain.enums.ReceiveTypeEnum;
import com.waves.task.domain.enums.TaskReceiveStatusEnum;
import com.waves.task.domain.enums.TaskTypeEnum;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Component
public class ReceiveAtomicStrategy extends AbstractReceiveStrategy {

    @Resource
    private TaskReceiveDao taskReceiveDao;

    @Resource
    private TaskDao taskDao;

    @Resource
    private LockService lockService;

    @Override
    ReceiveTypeEnum getReceiveTypeEnum() {
        return ReceiveTypeEnum.RECEIVE_ATOMIC;
    }

    @Override
    @Transactional
    public R<Object> receiveTask(Task task) {
        // 如果是求助贴，不走下面的逻辑
        if (Objects.equals(TaskTypeEnum.TASK_TYPE_HELP.getType(), task.getType())) {
            return super.saveTaskReceive(task);
        }
        String key = TaskRedisKey.getKey(TaskRedisKey.TASK_RECEIVE_USER, task.getId());
        // 已领取任务数量
        Long num = RedisUtils.sGetSetSize(key);
        // 任务数量
        Integer storeNum = task.getBountyNum() != null ? task.getBountyNum() : task.getHelpBountyNum();
        if (num >= storeNum) {
            // 批量插入只会执行一次，阻塞下一个请求进来批量插入
            Integer count = taskReceiveDao.getCountByTaskId(task.getId());
            AssertUtil.isTrue(count == 0, "任务已经全部领完！");
            lockService.executeWithLock(String.valueOf(task.getId()), 0, TimeUnit.SECONDS, () -> {
                // 有任务说明已经领完了
                Integer count2 = taskReceiveDao.getCountByTaskId(task.getId());
                AssertUtil.isTrue(count2 == 0, "任务已经全部领完！");

                // 因为没有加锁的情况下，redis存储的数量大概率是超过赏金设定数量
                // 因此添加领取者时，通过atomic原子类进行库存扣减，
                AtomicInteger atomic = new AtomicInteger(storeNum);
                List<TaskReceive> list = Collections.synchronizedList(new ArrayList<>());
                Set<String> strings = RedisUtils.sGet(key);
                // 存放多线程并发执行的结果集
                List<CompletableFuture<Void>> completableFutureList = new ArrayList<>();
                for (String info : strings) {
                    // 使用多线程+原子类，快速创建领取信息放入List，以及移除任务超领的用户
                    CompletableFuture<Void> run = CompletableFutureUtils.createRun(() -> {
                        String replaceInfo = info.replace("\"", "");
                        String[] split = replaceInfo.split(",");
                        // decrementAndGet作用就是原子性的-1，并返回-1后的值
                        if (atomic.decrementAndGet() >= 0) {
                            TaskReceive taskReceive = new TaskReceive()
                                    .setTaskId(task.getId())
                                    .setStatus(TaskReceiveStatusEnum.RECEIVE_STATUS_ING.getType())
                                    .setReceiverId(Long.parseLong(split[0]))
                                    .setReceiverName(split[1])
                                    .setReceiveTime(new Date().getTime());
                            list.add(taskReceive);
                        } else {
                            // 如果为0，就把集合多出来用户的移除
                            RedisUtils.lRemove(key, 1, info);
                        }
                    });
                    completableFutureList.add(run);
                }
                // 阻塞多线程，用来获取值
                CompletableFuture.allOf(completableFutureList.toArray(CompletableFuture[]::new)).join();
                // todo 用mq发送给websocket服务，推送给这些在redis集合中的用户一个领取任务成功的提示

                // 批量入库
                taskReceiveDao.saveAll(list);
                // 任务降级，有人取消任务，就不能再用这种方式领取任务；
                // 这种是用来处理超大库存量和并发量一抢而空的情况
                task.setLevel(ReceiveTypeEnum.RECEIVE_RLOCK.getType());
                taskDao.updateById(task);
                return R.ok();
            });
        } else {
            // 如果还有库存，将用户uid存入set集合
            User user = LoginUtils.getUser();
            RedisUtils.sSetAndTime(key, TimeConstant.YEAR_TIME, user.getId() + "," + user.getName());
            AssertUtil.isTrue(false, "正在疯狂抢单，请稍后");
        }
        // ①如果任务没有一抢而空，而是迟迟抢购不完，这样所有的领取者又无法入库，这种情况可以先将级别降低，
        // 然后用一个定时任务，比如任务开始后两秒内还没有抢完，就把已经抢的存在redis里的用户入库，剩下的任务慢慢用RLock锁策略领取
        return null;
    }

    @Override
    public void cancelTask(Task task) {

    }

    @Override
    public void createTask(Task task) {

    }

}
